package com.taobao.api.internal.stream;

import java.io.IOException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import com.founder.android.util.Logger;

import com.taobao.api.internal.stream.connect.ConnectionLifeCycleListener;
import com.taobao.api.internal.stream.connect.HttpClient;
import com.taobao.api.internal.stream.connect.HttpResponse;
import com.taobao.api.internal.stream.message.MessageStreamImpl;
import com.taobao.api.internal.stream.message.StreamMsgConsumeFactory;
import com.taobao.api.internal.stream.message.TopCometMessageListener;
import com.taobao.api.internal.util.RequestParametersHolder;
import com.taobao.api.internal.util.StringUtils;
import com.taobao.api.internal.util.TaobaoHashMap;
import com.taobao.api.internal.util.TaobaoUtils;

/**
 * 
 * @author zhenzi 2011-8-9 上午09:59:31
 */
public class TopCometStreamImpl implements TopCometStream {
	private static final Logger logger = Logger
			.getLogger(TopCometStreamImpl.class);
	private ConnectionLifeCycleListener connectionListener;
	private TopCometMessageListener cometMessageListener;
	private Configuration conf;
	private HttpClient httpClient;
	private StreamMsgConsumeFactory msgConsumeFactory = null;
	/**
	 * 停掉
	 */
	private boolean stop = false;
	/*
	 * 各种状态
	 */
	private String serverRespCode = StreamConstants.CLIENT_FIRST_CONNECT;
	private ReentrantLock lock = new ReentrantLock();
	private Condition controlCondition = lock.newCondition();

	private int startConsumeThreadTimes = 0;
	private long lastStartConsumeThread = System.currentTimeMillis();

	private StreamImplementation currentStreamImpl;

	protected TopCometStreamImpl(Configuration conf) {
		this.conf = conf;
	}

	public void setConnectionListener(
			ConnectionLifeCycleListener connectionLifeCycleListener) {
		this.connectionListener = connectionLifeCycleListener;
	}

	public void setMessageListener(TopCometMessageListener cometMessageListener) {
		this.cometMessageListener = cometMessageListener;
	}

	public void start() {
		if (cometMessageListener == null) {
			throw new RuntimeException("Comet message listener must not null");
		}
		TaobaoHashMap param = new TaobaoHashMap();
		param.put(StreamConstants.PARAM_APPKEY, conf.getAppkey());
		if (conf.getUserid() > 0) {
			param.put(StreamConstants.PARAM_USERID, conf.getUserid());
		}
		if (conf.getId() != null){
			param.put(StreamConstants.PARAM_CONNECT_ID, conf.getId());
		}
		String timestamp = String.valueOf(System.currentTimeMillis());
		param.put(StreamConstants.PARAM_TIMESTAMP, timestamp);
		RequestParametersHolder paramHolder = new RequestParametersHolder();
		paramHolder.setProtocalMustParams(param);
		String sign = null;
		try {
			sign = TaobaoUtils.signTopRequestNew(paramHolder, conf.getSecret(),
					false);
			if (StringUtils.isEmpty(sign)) {
				throw new RuntimeException("Get sign error");
			}
		} catch (IOException e) {
			throw new RuntimeException(e);
		}
		param.put(StreamConstants.PARAM_SIGN, sign);
		httpClient = new HttpClient(conf, param);
		msgConsumeFactory = new StreamMsgConsumeFactory(conf.getMinThreads(),
				conf.getMaxThreads(), conf.getQueueSize());

		new Thread(new ControlThread(), "stream-control-thread").start();
	}

	public StreamImplementation getMsgStreamImpl()
			throws TopCometSysErrorException {
		HttpResponse response;
		try {
			response = httpClient.post();
		} catch (TopCometSysErrorException e) {
			if (connectionListener != null) {
				connectionListener.onConnectError(e);
			}
			throw e;
		}
		return currentStreamImpl = new MessageStreamImpl(msgConsumeFactory,
				response, cometMessageListener, this);
	}

	class ControlThread implements Runnable {
		private static final String threadName = "top-stream-consume-thread";
		private TopCometStreamConsume currentStreamConsume;
		private boolean isReconnect = false;//是否客户端发起重连
		public void run() {
			long lastSleepTime = 0;
			while (!stop) {
				try {
					if (StreamConstants.SERVER_DEPLOY.equals(serverRespCode)) {// 服务端在发布
						if (logger.isDebugEnabled()) {
							logger.debug("Server is upgrade sleep "
									+ conf.getSleepTimeOfServerInUpgrade()
									+ " seconds");
						}
						try {
							Thread.sleep(conf.getSleepTimeOfServerInUpgrade() * 1000);
						} catch (InterruptedException e) {
							// ignore;
						}
						startConsumeThread();
					} else if (/*客户端第一次发起连接请求*/
							StreamConstants.CLIENT_FIRST_CONNECT.equals(serverRespCode) || 
							/*服务端主动断开了所有的连接*/
							StreamConstants.SERVER_REHASH.equals(serverRespCode) || 
							/*连接到达最大时间*/
							StreamConstants.CONNECT_REACH_MAX_TIME.equals(serverRespCode) ||
							/*在一些异常情况下需要重连*/
							StreamConstants.RECONNECT.equals(serverRespCode)) {
						startConsumeThread();
					} else if (/*客户端自己把自己踢开*/
							StreamConstants.CLIENT_KICKOFF.equals(serverRespCode) ||
							/*服务端把客户端踢开*/
							StreamConstants.SERVER_KICKOFF.equals(serverRespCode)) {
						if((StreamConstants.CLIENT_KICKOFF.equals(serverRespCode) && !isReconnect) ||
						   StreamConstants.SERVER_KICKOFF.equals(serverRespCode)){ 
							break;// 终止掉当前线程
						}
					}else{//错误码设置出错，停止线程
						stop = true;
						break;
					}
					//连接成功，开始休眠
					try{
						lock.lock();
						lastSleepTime = System.currentTimeMillis();
						controlCondition.await(conf.getHttpReconnectInterval(), TimeUnit.SECONDS);
						if(System.currentTimeMillis() - lastSleepTime >= conf.getHttpReconnectInterval() * 1000){
							/*
							 * 快要到达连接的最大时间了，需要重新发起连接
							 */
							serverRespCode = StreamConstants.RECONNECT;
							isReconnect = true;
						}//否则，是由于某种原因被唤醒的
					}catch(Exception e){
						logger.error(e, e);
					}finally{
						lock.unlock();
					}
				} catch (Throwable e) {
					logger.error("Occur some error,stop the stream consume", e);
					stop = true;
					try {
						lock.lock();
						controlCondition.signalAll();
					} catch (Exception ex) {
						// ignore
					} finally {
						lock.unlock();
					}
				}
			}
			if (currentStreamImpl != null) {
				try {
					currentStreamImpl.close();
				} catch (IOException e) {
					// ignore
				}
			}
			msgConsumeFactory.shutdown();
			logger.info("Stop stream consume");
		}

		private void startConsumeThread() {
			currentStreamConsume = new TopCometStreamConsume();
			Thread consumeThread = new Thread(currentStreamConsume, threadName);
			consumeThread.setDaemon(true);
			consumeThread.start();
			lastStartConsumeThread = System.currentTimeMillis();
		}

	}

	class TopCometStreamConsume implements Runnable {
		private StreamImplementation stream;
		private boolean closed = false;
		
		TopCometStreamConsume() {
		}

		public void run() {
			boolean connected = false;
			if (!closed && null == stream) {
				if (logger.isDebugEnabled()) {
					logger.debug("Establishing connection.");
				}
				try {
					stream = getStream();
				} catch (TopCometSysErrorException sysErrorException) {// 在发起连接的时候发生了异常，退出
					stop = true;
					closed = true;
					if (connectionListener != null) {
						try {// 避免实现者处理的时候抛出异常导致不能正常执行接下来的退出逻辑
							connectionListener
									.onSysErrorException(sysErrorException);
						} catch (Exception e) {
							logger.error(e, e);
						}
					}
					return;
				}
				connected = true;
				if (logger.isDebugEnabled()) {
					logger.info("Connection established.");
				}
				if (connectionListener != null) {
					try {// 避免实现者处理的时候抛出异常导致不能正常执行接下来的逻辑
						connectionListener.onConnect();
					} catch (Exception e) {
						logger.warn(e.getMessage());
					}
				}
				while (!closed && stream.isAlive()) {
					try {
						stream.nextMsg();
					} catch (IOException e) {//出现了read time out异常
						// 资源清理
						if (stream != null) {
							try {
								stream.close();
							} catch (IOException e1) {
								logger.error(e1, e1);
							}
						}
						stream = null;
						connected = false;
						closed = true;
						//通知
						if (connectionListener != null) {
							try {
								connectionListener.onReadTimeout();
							} catch (Exception ex) {
								logger.error(ex, ex);
							}
						}
						/**
						 * 30分钟内发送了10次IOException
						 */
						if (System.currentTimeMillis()
								- lastStartConsumeThread < 30 * 60 * 1000) {// 短时间内由于读取IOException连接了10次，则退出
							startConsumeThreadTimes++;
							if (startConsumeThreadTimes >= 10) {
								stop = true;
								if (connectionListener != null) {
									try{
										connectionListener
											.onMaxReadTimeoutException();
									}catch(Exception maxE){logger.error(maxE, maxE);}
								}
								logger.error("Occure too many exception,stop the system,please check");
								//通知唤醒控制线程，但是不在发起重连接
								try {
									lock.lock();
									controlCondition.signalAll();
								} catch (Exception e2) {
								} finally {
									lock.unlock();
								}
							}else{//没有到达10次，通知重连
								startConsumeThreadTimes = 0;
								serverRespCode = StreamConstants.RECONNECT;
								try {
									lock.lock();
									controlCondition.signalAll();
								} catch (Exception e2) {
								} finally {
									lock.unlock();
								}
							}
						} else {
							// 通知重连
							startConsumeThreadTimes = 0;
							serverRespCode = StreamConstants.RECONNECT;
							try {
								lock.lock();
								controlCondition.signalAll();
							} catch (Exception e2) {
							} finally {
								lock.unlock();
							}
						}
					}
				}
			}
			//出现异常情况下做资源清理
			if (this.stream != null && connected) {
				try {
					this.stream.close();
				} catch (Exception e) {
					logger.warn(e.getMessage(), e);
				}
			}
		}

		StreamImplementation getStream() throws TopCometSysErrorException {
			return getMsgStreamImpl();
		}
	}

	public void stop() {
		stop = true;
		try {
			lock.lock();
			controlCondition.signalAll();
		} catch (Exception e) {
		} finally {
			lock.unlock();
		}
	}

	public ReentrantLock getLock() {
		return lock;
	}

	public Condition getControlCondition() {
		return controlCondition;
	}

	public void setServerRespCode(String serverRespCode) {
		this.serverRespCode = serverRespCode;
	}

}
